package org.zjt.mr.wordcnt;


import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FSDataOutputStream;
import org.apache.hadoop.fs.FileStatus;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;

import java.io.BufferedReader;
import java.io.FileInputStream;
import java.io.InputStreamReader;
import java.net.URI;

public class RunJob {

    public static void main(String[] args) throws Exception {
        updateDir();
        executeJob();
        printResult();
    }


    public static void executeJob() {
        Configuration config = new Configuration();
        config.set("fs.defaultFS", "hdfs://master:9000");
        config.set("yarn.resourcemanager.hostname", "master");
        System.setProperty("HADOOP_USER_NAME", "root");
//		config.set("mapred.jar", "C:\\Users\\Administrator\\Desktop\\wc.jar");
        try {
            FileSystem fs = FileSystem.get(config);

            Job job = Job.getInstance(config);
            job.setJarByClass(RunJob.class);

            job.setJobName("wc");

            job.setMapperClass(WordCountMapper.class);
            job.setReducerClass(WordCountReducer.class);

            job.setMapOutputKeyClass(Text.class);
            job.setMapOutputValueClass(IntWritable.class);

            FileInputFormat.addInputPath(job, new Path("/usr/input/"));

            Path outpath = new Path("/usr/output/wc");
            if (fs.exists(outpath)) {
                fs.delete(outpath, true);
            }
            FileOutputFormat.setOutputPath(job, outpath);

            boolean f = job.waitForCompletion(true);
            if (f) {
                System.out.println("job 运行完毕");
            }
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    public static void updateDir() throws Exception {
        Configuration config = new Configuration();
        FileSystem fs = FileSystem.get(new URI("hdfs://master:9000/"), config, "root");
        Path file = new Path("/usr/input/wc.txt");
        FSDataOutputStream out = fs.create(file);
//		IOUtils.copy(input, output)
        org.apache.hadoop.io.IOUtils.copyBytes(new FileInputStream("/Users/zhangjuntao/IdeaProjects/myproject/mapreducer/data/wc.txt"), out, config);
        fs.close();
    }


    public static void printResult() throws Exception {
        Configuration config = new Configuration();
        FileSystem fs = FileSystem.get(new URI("hdfs://master:9000/"), config, "root");
        Path file = new Path("/usr/output/wc/");
        FileStatus[] fss = fs.listStatus(file);
        for (FileStatus fileName : fss) {
            System.out.println(fileName.getPath().getName() + ":");
            BufferedReader br = new BufferedReader(new InputStreamReader(fs.open(fileName.getPath())));
            String line;
            line = br.readLine();
            while (line != null) {
                System.out.println("\t" + line);
                line = br.readLine();
            }
        }
        fs.close();
    }
}
